package com.xzx.flink.streamapi.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * bin/zkServer.sh --config conf start
 * bin/kafka-server-start.sh config/server.properties
 * bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic sensor
 * bin/kafka-console-producer.sh --broker-list localhost:9092 --topic sensor
 * @author xinzhixuan
 * @version V1.0.0
 * @date 2021/4/20
 **/
public class SourceFromKafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // kafka 配置项
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test");
//        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");
        // 从 kafka 读取数据
        DataStream<String> source = env.addSource(new FlinkKafkaConsumer<>("sensor", new SimpleStringSchema(), properties));

        source.print();

        env.execute();
    }
}
